package com.atguigu.gulimall.seckill.service.impl;

import com.alibaba.csp.sentinel.Entry;
import com.alibaba.csp.sentinel.SphU;
import com.alibaba.csp.sentinel.annotation.SentinelResource;
import com.alibaba.csp.sentinel.slots.block.BlockException;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.atguigu.common.to.mq.SeckillOrderTo;
import com.atguigu.common.utils.R;
import com.atguigu.common.vo.MemberResponseVo;
import com.atguigu.gulimall.seckill.fegin.CouponFeignService;
import com.atguigu.gulimall.seckill.fegin.ProductFeignService;
import com.atguigu.gulimall.seckill.interceptor.LoginUserInterceptor;
import com.atguigu.gulimall.seckill.service.SeckillService;
import com.atguigu.gulimall.seckill.to.SeckillSkuRedisTo;
import com.atguigu.gulimall.seckill.vo.SeckillSessionWithSkus;
import com.atguigu.gulimall.seckill.vo.SkuInfoVo;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RSemaphore;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import javax.xml.crypto.Data;
import java.util.Date;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

@Slf4j
@Service
public class SeckillServiceImpl implements SeckillService {
    @Autowired CouponFeignService couponFeignService;
    @Autowired StringRedisTemplate redisTemplate;
    @Autowired ProductFeignService productFeignService;
    @Autowired RedissonClient redissonClient;
    @Autowired RabbitTemplate rabbitTemplate;
    private final String SESSIONS_CACHE_PREFIX = "seckill:sessions:";    //缓存秒杀的场次的key
    private final String SKUKILL_CACHE_PREFIX = "seckill:skus:";    //缓存秒杀的商品的key
    private final String SKU_STOCK_SEMAPHORE = "seckill:stock:";    //作为信号量用

    /**
     * 秒杀抢购
     * 整个方法都是在缓存中操作，速度非常快
     */
    @Override
    public String kill(String killId, String key, Integer num) {
        long s1 = System.currentTimeMillis();
        //1、从拦截器获取用户信息
        MemberResponseVo repsVo = LoginUserInterceptor.loginUser.get();

        //2、获取当前商品的详细信息，hash 操作，从 SKUKILL_CACHE_PREFIX 这个 key 中找到所有参与秒杀的商品的信息
        BoundHashOperations<String, String, String> hashOps = redisTemplate.boundHashOps(SKUKILL_CACHE_PREFIX);
        String json = hashOps.get(killId);    //根据 key 在 SKUKILL_CACHE_PREFIX 中获取值

        if (!StringUtils.isEmpty(json)){
            SeckillSkuRedisTo redisTo = JSON.parseObject(json, SeckillSkuRedisTo.class);

            //3、校验合法性
            Long startTime = redisTo.getStartTime();    //开始时间
            Long endTime = redisTo.getEndTime();    //结束时间
            long current = new Date().getTime();    //当前时间
            long ttl = endTime - startTime;    //场次存活时间

            //3.1 校验时间的合法性
            if (current >= startTime && current < endTime){
                //3.2 校验随机码和商品id
                String randomCode = redisTo.getRandomCode();    //商品秒杀的随机码
                String skuId = redisTo.getPromotionSessionId() + "_" + redisTo.getSkuId();    //活动场次id + "_" + 商品id

                if (randomCode.equals(key) && skuId.equals(killId)){
                    //3.3 验证秒杀的数量是否合理
                    if (num <= redisTo.getSeckillLimit()){    //每人限购数量
                        //3.4 验证这个人是否购买过。幂等性处理。如果只要秒杀成功，就去占位  userId_sessionId_skillId
                        String redisKey = repsVo.getId() + "_" + skuId;
                        //3.5 自动过期--通过在redis中使用 用户id-skuId 来占位看是否买过，setIfAbsent：如果不存在则设置
                        //                                                           key       value       存活时间      单位毫秒
                        Boolean ifAbsent = redisTemplate.opsForValue().setIfAbsent(redisKey, num.toString(), ttl, TimeUnit.MILLISECONDS);
                        if (ifAbsent){
                            //3.6 占位成功，说明该用户未秒杀过该商品，则继续尝试获取库存信号量，(之前定义的：信号量就是商品秒杀总量)
                            RSemaphore semaphore = redissonClient.getSemaphore(SKU_STOCK_SEMAPHORE + randomCode);
                            boolean b = semaphore.tryAcquire(num);
                            if (b){
                                //秒杀成功
                                String timeId = IdWorker.getTimeId();
                                SeckillOrderTo orderTo = new SeckillOrderTo();
                                orderTo.setOrderSn(timeId);    //订单号
                                orderTo.setMemberId(repsVo.getId());    //会员id
                                orderTo.setNum(num);    //秒杀数量
                                orderTo.setPromotionSessionId(redisTo.getPromotionSessionId());    //秒杀场次id
                                orderTo.setSkuId(redisTo.getSkuId());    //商品id
                                orderTo.setSeckillPrice(redisTo.getSeckillPrice());    //秒杀价格
                                //TODO 发送 MQ 消息，使用队列，做流量削峰
                                rabbitTemplate.convertAndSend("order-event-exchange","order.seckill.order", orderTo);
                                long s2 = System.currentTimeMillis();
                                log.info("耗时..." + (s2 - s1));
                                return timeId;
                            }
                            return null;
                        }else {
                            //3.7 说明已经买过
                            return null;
                        }
                    }
                }else {
                    return null;
                }
            }else {
                return null;
            }
        }
        return null;
    }


    /**
     * 查询当前商品是否参与秒杀活动
     */
    @Override
    public SeckillSkuRedisTo getSkuSecKillInfo(Long skuId) {
        //1、从 SKUKILL_CACHE_PREFIX 这个 key 中找到所有需要参与秒杀的商品的 key
        BoundHashOperations<String, String, String> hashOps = redisTemplate.boundHashOps(SKUKILL_CACHE_PREFIX);
        Set<String> keys = hashOps.keys();
        if (keys != null && keys.size() > 0){
            //2、判断当前 key 是否匹配
            String regx = "\\d_" + skuId;    //正则表达式，"\\d_"：d 表示一个数字，只要是一个数字加 "_"，就是匹配的，比如 6_4，就是匹配的
            for (String key : keys) {
                //3、如果缓存中的 key 和传过来的 key 匹配
                if (Pattern.matches(regx, key)){
                    //4、从 SKUKILL_CACHE_PREFIX 这个 key 中找到指定 key 的值
                    String json = hashOps.get(key);
                    SeckillSkuRedisTo skuRedisTo = JSON.parseObject(json, SeckillSkuRedisTo.class);

                    //5、随机码，判断传过来的商品是否在秒杀活动的时间范围内
                    Long current = new Date().getTime();
                    if (current >= skuRedisTo.getStartTime() && current < skuRedisTo.getEndTime()){
                        //说明当前秒杀场次在秒杀的时间范围内
                        return skuRedisTo;
                    }else {
                        skuRedisTo.setRandomCode(null);    //不在秒杀的时间范围内，把随机码设为 null
                    }
                    return skuRedisTo;
                }
            }
        }
        return null;
    }

    /**
     * 处理 @SentinelResource 标注的方法的降级
     * 返回值类型要和受保护方法的返回值一样
     */
    public List<SeckillSkuRedisTo> blockHandler(BlockException e){
        log.error("getCurrentSeckillSkusResource...被限流了...");
        return null;
    }

    /**
     * 查询当前时间可以参与的秒杀商品信息
     */
    //value：受保护资源的名字, blockHandler：当被降级以后，被哪个方法处理，fallback：还可以设置 fallback 函数，会针对所有类型的异常
    @SentinelResource(value = "getCurrentSeckillSkusResource", blockHandler = "blockHandler")
    @Override
    public List<SeckillSkuRedisTo> getCurrentSeckillSkus() {
        Long time = new Date().getTime();

        //TODO sentinel 自定义资源保护
        try(Entry entry = SphU.entry("seckillSkus")){
            //TODO try 里放受保护的资源
            //1、获取所有 SESSIONS_CACHE_PREFIX 类型的 keys (秒杀场次的key)
            Set<String> keys = redisTemplate.keys(SESSIONS_CACHE_PREFIX + "*");

            for (String key : keys) {
                //2、拿到每个秒杀场次的key，缓存中的键值的格式 => seckill:sessions:1672855200000_1672862400000
                String replace = key.replace(SESSIONS_CACHE_PREFIX, "");    //把 SESSIONS_CACHE_PREFIX 替换成 ""，得到后面的时间值
                String[] split = replace.split("_");    //把 1672855200000_1672862400000 以 "_" 进行分割
                long startTime = Long.parseLong(split[0]);    //拿到第一个时间作为开始时间
                long endTime = Long.parseLong(split[1]);    //拿到第二个时间作为结束时间

                //3、判断当前秒杀活动是否处于有效期内
                if (time > startTime && time < endTime) {
                    //4、根据当前秒杀场次的 key 获取所有商品信息，range：获取区间值，区间范围：-100 - 100
                    List<String> range = redisTemplate.opsForList().range(key, -100, 100);

                    //5、hash 操作，针对 SKUKILL_CACHE_PREFIX 这个 key 的操作
                    BoundHashOperations<String, String, String> hashOps = redisTemplate.boundHashOps(SKUKILL_CACHE_PREFIX);
                    assert range != null;
                    List<String> list = hashOps.multiGet(range);    //multiGet：获取多个 key 的数据
                    if (!CollectionUtils.isEmpty(list)) {
                        List<SeckillSkuRedisTo> collect = list.stream().map(item -> JSON.parseObject(item, SeckillSkuRedisTo.class))
                                .collect(Collectors.toList());
                        return collect;
                    }
                    break;
                }
            }
        }catch (BlockException e){
            log.error("资源被限流：{}", e.getMessage());
        }
        return null;
    }

    /**
     * 定时上架最近三天需要秒杀的商品
     */
    @Override
    public void uploadSeckillSkuLatest3Days() {
        //1、扫描最近三天需要参与秒杀的活动
        R session = couponFeignService.getLasts3DaySession();    //远程调用《获取最近三天的秒杀活动 以及 关联的秒杀商品》方法
        if (session.getCode() == 0) {
            //2、上架商品
            List<SeckillSessionWithSkus> data = session.getData(new TypeReference<List<SeckillSessionWithSkus>>() {});
            //3、缓存到redis
            saveSessionInfos(data);    //调用《保存秒杀场次信息到 redis 中》方法
            saveSessionSkuInfos(data);    //调用《保存秒杀商品信息到 redis 中》方法
        }
    }


    /**
     * 保存秒杀场次信息到 redis 中
     */
    private void saveSessionInfos(List<SeckillSessionWithSkus> sessions) {
        if (!CollectionUtils.isEmpty(sessions)) {
            sessions.stream().forEach(session -> {
                long startTime = session.getStartTime().getTime();    //每日开始时间, 转成 Long 类型，便于后面比较时间
                long endTime = session.getEndTime().getTime();    //每日结束时间, 转成 Long 类型，便于后面比较时间

                String key = SESSIONS_CACHE_PREFIX + startTime + "_" + endTime;    //SESSIONS_CACHE_PREFIX = "seckill:sessions:"
                //TODO 幂等性处理，判断缓存中是否有指定的 key
                Boolean hasKey = redisTemplate.hasKey(key);
                //如果没有当前 key，才缓存场次信息
                if (!hasKey) {
                    List<String> collect = session.getRelationSkus().stream()
                                               //获取活动场次id + "_" + 商品id
                            .map(item -> item.getPromotionSessionId().toString() + "_" + item.getSkuId().toString())
                            .collect(Collectors.toList());
                    //缓存活动信息
                    redisTemplate.opsForList().leftPushAll(key, collect);    //leftPushAll: 从左往右保存整个集合
                }});}
    }

    /**
     * 保存秒杀商品信息到 redis 中
     */
    private void saveSessionSkuInfos(List<SeckillSessionWithSkus> sessions) {
        //1、准备 hash 操作，boundHashOps：在指定的 key(SKUKILL_CACHE_PREFIX) 中操作        SKUKILL_CACHE_PREFIX = "seckill:skus:"
        BoundHashOperations<String, Object, Object> ops = redisTemplate.boundHashOps(SKUKILL_CACHE_PREFIX);

        sessions.stream().forEach(session -> {
            session.getRelationSkus().stream().forEach(seckillSkuVo -> {
                //TODO 幂等性处理，判断 SKUKILL_CACHE_PREFIX 这个 key 里面，是否有指定 key，没有才能缓存
                if (!ops.hasKey(seckillSkuVo.getPromotionSessionId() + "_" + seckillSkuVo.getSkuId().toString())){
                    //缓存商品
                    SeckillSkuRedisTo redisTo = new SeckillSkuRedisTo();
                    //2、sku的基本信息
                    R r = productFeignService.getSkuInfo(seckillSkuVo.getSkuId());    //远程调用《根据 skuId 查询 sku 信息》方法
                    if (0 == r.getCode()) {
                        SkuInfoVo skuInfo = r.getData("skuInfo", new TypeReference<SkuInfoVo>() {});
                        redisTo.setSkuInfo(skuInfo);
                    }
                    //3、sku的秒杀信息
                    BeanUtils.copyProperties(seckillSkuVo, redisTo);

                    //4、设置当前商品的秒杀时间信息
                    redisTo.setStartTime(session.getStartTime().getTime());
                    redisTo.setEndTime(session.getEndTime().getTime());

                    //5、随机码
                    String token = UUID.randomUUID().toString().replace("_", "");
                    redisTo.setRandomCode(token);

                    String jsonString = JSON.toJSONString(redisTo);
                    //6、把键值数据存到 SKUKILL_CACHE_PREFIX 这个 key 中
                    ops.put(seckillSkuVo.getPromotionSessionId() + "_" + seckillSkuVo.getSkuId().toString(), jsonString);

                    //7、使用库存作为分布式信号量 限流                         SKU_STOCK_SEMAPHORE = "seckill:stock:"
                    RSemaphore semaphore = redissonClient.getSemaphore(SKU_STOCK_SEMAPHORE + token);
                    semaphore.trySetPermits(seckillSkuVo.getSeckillCount());    //使用商品秒杀总量作为信号量
                }});});
    }
}
